package cool.taomu.toolkit.service.centre.kademlia.service

import com.google.common.base.Optional
import com.google.common.util.concurrent.Futures
import com.google.common.util.concurrent.ListenableFuture
import com.google.common.util.concurrent.ListeningExecutorService
import com.google.common.util.concurrent.RateLimiter
import com.google.inject.Inject
import com.google.inject.name.Named
import cool.taomu.toolkit.impl.StoreKeyValue
import cool.taomu.toolkit.inter.IStoreKeyValue
import cool.taomu.toolkit.jersey.JerseyInternalInvoker
import cool.taomu.toolkit.service.centre.kademlia.entity.KResult
import cool.taomu.toolkit.service.centre.kademlia.utils.ResourceCollector
import cool.taomu.toolkit.service.kademlia.entity.FindNodeEntity
import cool.taomu.toolkit.service.kademlia.entity.FindServiceNodeEntity
import cool.taomu.toolkit.service.kademlia.entity.KPeerNode
import cool.taomu.toolkit.service.kademlia.entity.RequestDto
import cool.taomu.toolkit.service.kademlia.service.ICheckPeer
import cool.taomu.toolkit.service.kademlia.service.KRoutingTable
import cool.taomu.toolkit.service.kademlia.utils.KademliaUtils
import cool.taomu.toolkit.utils.PeerUtils
import java.util.ArrayList
import java.util.List
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException
import java.util.function.Function
import java.util.stream.Collectors
import org.apache.commons.beanutils.BeanUtils
import org.slf4j.LoggerFactory

class ServiceCentre {
    val static LOG = LoggerFactory.getLogger(ServiceCentre);

    @Inject(optional=true)
    IStoreKeyValue<String, byte[]> store = new StoreKeyValue<String, byte[]>();

    @Inject
    KRoutingTable router;

    @Inject
    ListeningExecutorService executor;

    @Inject
    ICheckPeer checkPeer;

    @Inject
    @Named("taomu.toolkit.kademlia.findNode")
    Function<FindNodeEntity, List<KPeerNode>> findNode

    @Inject
    FindNodeService fnodeService;

    @Deprecated
    def Optional<List<KPeerNode>> findNode(String nodeId, List<String> excludeIds) {
        LOG.info("****findNode****:{}", nodeId);
        var findNodes = router.findClosestNodes(nodeId);
        LOG.info("查找到的节点数量:{}", findNodes.size);
        val ListenableFutures = new ArrayList<ListenableFuture<List<KPeerNode>>>();
        LOG.info("查找的id:{}", nodeId);
        val resultNodes = newArrayList();
        excludeIds.add(router.localNode.id)
        findNodes.filter [
            !excludeIds.contains(it.id)
        ].forEach [ node |
            resultNodes.add(node)
            var futures = executor.submit([
                LOG.info("连接远程服务器 {} 查找节点", node);
                return findNode.apply(new FindNodeEntity(nodeId, node, router.localNode, excludeIds))
            ])
            ListenableFutures.add(futures);
        ]
        return this.getFindNodes(ListenableFutures, resultNodes, nodeId);
    }

    @Deprecated
    private def getFindNodes(ArrayList<ListenableFuture<List<KPeerNode>>> ListenableFutures,
        ArrayList<KPeerNode> resultNodes, String nodeId) {
        LOG.info("执行查找节点的线程数:{}", ListenableFutures.size);
        var ListenableFuture<List<List<KPeerNode>>> allFutures = Futures.allAsList(ListenableFutures);
        var List<List<KPeerNode>> results;
        try {
            // 设置最大等待时间（例如 5 秒）
            results = allFutures.get(5, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            LOG.info("部分节点调用超时，尝试继续处理已有结果");
            results = Futures.successfulAsList(ListenableFutures).get(); // 获取已成功的结果
        }
        var closestNodes = results.stream().filter[it !== null] // 过滤掉 null 的 List
        .flatMap[it.stream()] // 安全 flatMap
        .filter[it !== null] // 过滤掉可能为 null 的 KPeerNode
        .distinct.collect(Collectors.toList())
        LOG.info("closestNodes 返回节点数:{}", closestNodes.size);
        resultNodes.addAll(closestNodes);
        if (resultNodes.size == 0) {
            return Optional.absent();
        }
        LOG.info("需要将网络上找到的{}个节点最新文件下载到本地并加到桶中", closestNodes.size);
        closestNodes.forEach [
            executor.execute([
                this.checkPeer.apply(it)
            // router.addPeerNode(it, checkPeer);
            ])
        ]
        var temp = resultNodes.toSet.toList
        resultNodes.clear
        resultNodes.addAll(temp);
        LOG.info("resultNodes 中的节点数 {} 排序找到的节点", resultNodes.size);
        resultNodes.sort([ n1, n2 |
            PeerUtils.xorDistance(nodeId, n1.getId()).compareTo(PeerUtils.xorDistance(nodeId, n2.getId()))
        ])
        var nodes = resultNodes.subList(0, Math.min(router.k, resultNodes.size()));
        return Optional.fromNullable(nodes)
    }

    @Inject
    ICheckRemoteNode crn;

    /**
     * peer 为远程端 节点信息
     */
    def Optional<String> ping(KPeerNode peer) {
        if (!peer.host.equalsIgnoreCase("localhost")) {
            this.executor.execute([
                if (!crn.bq.contains(peer)) {
                    crn.bq.add(peer)
                }
            ])
        }
        return Optional.fromNullable("PONG");
    }

    def Optional<KResult<String>> findValue(String key) {
    }

    def Optional<Boolean> store(String key, String value) {
        store.put(key, value.bytes)
        return Optional.absent;
    }

    @Inject
    RateLimiter rateLimiter;

    @Inject
    ResourceCollector collector

    @Inject
    JerseyInternalInvoker invoker;

    @Inject
    @Named("taomu.toolkit.kademlia.findServiceNode")
    Function<FindServiceNodeEntity, List<KPeerNode>> findServiceNodeFunc

    def Optional<List<KPeerNode>> findServiceNode(RequestDto dto, List<String> excludeIds) {
        val key = #[dto.method, dto.serviceName].join(":");
        var boolean tryAcquire = rateLimiter.tryAcquire(500, TimeUnit.MILLISECONDS);
        val results = newArrayList();
        if (tryAcquire) {
            if (collector.matchesAnyTemplate(key)) {
                results.add(this.router.localNode)
            // return Optional.fromNullable(results);
            }
        } else {
            LOG.info("限流超时了:{}", dto);
        }
        excludeIds.add(this.router.localNode.id)
        var key160 = KademliaUtils.stringTo160Bit(key);
        var nodes = fnodeService.findNode(key160, excludeIds)
        val ListenableFutures = new ArrayList<ListenableFuture<List<KPeerNode>>>();
        if (nodes.present) {
            nodes.get.filter [
                !excludeIds.contains(it.id)
            ].forEach [
                var futures = executor.submit([
                    findServiceNodeFunc.apply(new FindServiceNodeEntity(dto, it, this.router.localNode, excludeIds))
                ])
                ListenableFutures.add(futures);
            ]
        }
        // val resultNodes = newArrayList();
        return this.getFindNodes(ListenableFutures, results, key160);
    }

    def Optional<String> callService(RequestDto dto) {
        var key = #[dto.method, dto.serviceName].join(":");
        LOG.info("***Key****:{}", key)
        var boolean tryAcquire = rateLimiter.tryAcquire(500, TimeUnit.MILLISECONDS);
        var key160 = KademliaUtils.stringTo160Bit(key);
        LOG.info("****key160****:{}", key160)
        if (collector.matchesAnyTemplate(key)) {
            LOG.info("****返回findService执行的结果1****")
            if (tryAcquire) {
                var idto = new cool.taomu.toolkit.jersey.RequestDto()
                BeanUtils.copyProperties(idto, dto)
                var ins = invoker.invoke(idto);
                if (ins.present) {
                    LOG.info("invoker 返回值: {}", ins.get())
                }
                return ins;
            } else {
                LOG.info("限流超时了:{}", dto);
            }
        }
        LOG.info("未找到服务：{}", key);
        return Optional.absent;
    }

}
